#include "XBaseClient.h"
#include "XBaseServer.h"
#include "XBaseCommand.h"
#include "XBaseError.h"
#include "XUtility.h"

#include <assert.h>

#ifdef XLOGGER
#include <log4cplus/configurator.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <fstream>
#endif

using namespace std;

XIMPL_LOGGER(XBaseClient, logger);

const uint32_t XBaseClient::COMPRESS_NONE = 0X0;
const uint32_t XBaseClient::COMPRESS_QUICK_LZ = qkLZ;
const uint32_t XBaseClient::COMPRESS_ZLIB = zlib;
const uint32_t XBaseClient::COMPRESS_MLZO = mLZO;

XBaseClient::XBaseClient(const char* ip_address, int port)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    MAX_BUFFER_SIZE = 3 * 1024 * 1024;
    mBuffer = new char[MAX_BUFFER_SIZE];

    mRecvTimeoutMs = mDefaultRecvTimeoutMs;
    mSendTimeoutMs = mDefaultSendTimeoutMs;
    
    mMaster = new XBaseServer;
    mTabletList = new XBaseServerManager;
    mRootTablet = new XBaseServer;

    mMaster->init(ip_address, port, mRecvTimeoutMs, mSendTimeoutMs);	
}
XBaseClient::XBaseClient(const std::string& ip_address, short port)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    MAX_BUFFER_SIZE = 3 * 1024 * 1024;
    mBuffer = new char[MAX_BUFFER_SIZE];

    mRecvTimeoutMs = mDefaultRecvTimeoutMs;
    mSendTimeoutMs = mDefaultSendTimeoutMs;

    mMaster = new XBaseServer;
    mTabletList = new XBaseServerManager;
    mRootTablet = new XBaseServer;

    mMaster->init(ip_address, port, mRecvTimeoutMs, mSendTimeoutMs);	
}
XBaseClient::~XBaseClient()
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    mMaster->close();
    mRootTablet->close();

    mTabletList->close();

    delete[] mBuffer;

    delete mMaster;
    delete mTabletList;
    delete mRootTablet;
}

void XBaseClient::set_recv_timeout(uint32_t timeoutMs) 
{ 
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    XLOG4CPLUS_DEBUG( logger, "INPUT: timeoutMs=" << timeoutMs );

    mRecvTimeoutMs = timeoutMs;
    mMaster->set_recv_timeout( mRecvTimeoutMs );
}
void XBaseClient::set_send_timeout(uint32_t timeoutMs)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    XLOG4CPLUS_DEBUG( logger, "INPUT: timeoutMs=" << timeoutMs );

    mSendTimeoutMs = timeoutMs;
    mMaster->set_send_timeout( mSendTimeoutMs );
}

XBaseServer* XBaseClient::get_master()
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if (!mMaster->is_connected()) {
        if (!mMaster->reconnect()) {
            mErrno = XERROR_SERVER_CONNECT;
            mErrstr = mMaster->errorstr();

            return NULL;
        }
    }

    return mMaster;
}
XBaseServer* XBaseClient::get_tablet(const std::string& table_name, const std::string& row)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    XBaseServer* tablet = mTabletList->get(table_name, row);
    if ( tablet == NULL ) {
        tablet = load_tablet(table_name, row);
    }

    if ( tablet != NULL && !tablet->is_connected()) {
        if (!tablet->reconnect()) {			
            mErrno = XERROR_SERVER_CONNECT;
            mErrstr = tablet->errorstr();

            mTabletList->remove(*tablet);

            return NULL;
        }
    }

    return tablet;
}
XBaseServer* XBaseClient::get_tablet(XBaseCommandBase* cmd)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    const std::string& table_name = cmd->table_name();
    const std::string& row = cmd->row();
    XBaseServer* tablet = get_tablet(table_name, row);	
    if ( tablet != NULL ) {
        cmd->tablet_name() = tablet->tablet_name();
    } else {
        XLOG4CPLUS_DEBUG( logger, "get_tablet() NULL: errno=" << mErrno << ": " << mErrstr );

        cmd->result( XERROR_FAIL );
        cmd->errorno( mErrno );
        cmd->errorstr( mErrstr );
    }

    return tablet;
}

XBaseServer* XBaseClient::get_root_tablet()
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if (!mRootTablet->is_connected()) {		
        if ( load_root_tablet() == NULL ) {	
            return NULL;
        }

        if (!mRootTablet->reconnect()) {
            mErrno = XERROR_SERVER_CONNECT;
            mErrstr = mRootTablet->errorstr();

            return NULL;
        }
    }

    return mRootTablet;
}

int XBaseClient::scan_open(XBaseCommandScan * cmd)
{	
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if ( cmd == NULL ) {
        return XERROR_SUCC;
    }
    XLOG4CPLUS_DEBUG( logger, "INPUT: cmd->table_name()=" << cmd->table_name()
        << ",HEX(cmd->_start_row)=" << XUtility::str_2_hex( cmd->start_row() )
        << ",HEX(cmd->_end_row)=" << XUtility::str_2_hex( cmd->end_row() ) );

    if ( !cmd->end_row().empty() && cmd->start_row() >= cmd->end_row() ) {
        cmd->result( PERR_END_OF_SCAN_RECORD );
        cmd->errorno( PERR_END_OF_SCAN_RECORD );
        return XERROR_SUCC;
    }

    XBaseServer* tablet = get_tablet( cmd );
    if ( tablet == NULL ) {
        return XERROR_FAIL;
    }

    int res = scan_open( cmd,  tablet );
    if ( cmd->eof() && cmd->record().empty() 
        && !tablet->tablet_info().end_row().empty() 
        && ( cmd->end_row().empty() ||  tablet->tablet_info().end_row() < cmd->end_row()))
    {

        scan_close( cmd );
        if ( cmd->fail() ) {
            return cmd->result();
        }

        cmd->row() = tablet->tablet_info().end_row();
        scan_open( cmd );
        if ( cmd->eof() ) {
            return XERROR_SUCC;
        }
    }

    return res;
}

int XBaseClient::scan_close(XBaseCommandScan * cmd)
{	
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if (cmd == NULL) {
        return XERROR_SUCC;
    }	

    return scan_close( cmd,  get_tablet(cmd) );	
}

int XBaseClient::scan_next(XBaseCommandScan * cmd)
{	
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if (cmd == NULL) {
        return XERROR_SUCC;
    }

    return scan_next( cmd,  get_tablet(cmd) );	
}

int XBaseClient::scan_open(XBaseCommandScan * cmd, XBaseServer* tablet)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if ( cmd == NULL || tablet == NULL ) {
        return XERROR_SUCC;
    }

    cmd->open();

    std::string end_row = cmd->end_row();
    if ( !tablet->row_to().empty() && tablet->row_to() < cmd->end_row() ) {
        cmd->end_row() = tablet->row_to();
    }
    execute(cmd, tablet);
    cmd->end_row() = end_row;

    if ( tablet != mRootTablet && cmd->fatal() ) {
        XLOG4CPLUS_DEBUG(logger, "remove tablet");
        mTabletList->remove(*tablet);
    }

    if ( cmd->XBaseCommandBase::fail() ) {
        cmd->is_alive( false );
    } else {
        cmd->is_alive( true );
        cmd->start_row() = tablet->row_to();
    }

    return cmd->result();	
}

int XBaseClient::scan_close(XBaseCommandScan * cmd, XBaseServer* tablet)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if ( cmd == NULL || tablet == NULL ) {
        return XERROR_SUCC;
    }

    cmd->close();

    if ( cmd->is_alive() == false ) {
        cmd->result( XERROR_SUCC );
        return XERROR_SUCC;
    }
    cmd->is_alive( false );

    execute(cmd, tablet);

    if ( tablet != mRootTablet && cmd->fatal() ) {
        XLOG4CPLUS_DEBUG(logger, "remove tablet");
        mTabletList->remove(*tablet);
    }

    return cmd->result();	
}

int XBaseClient::scan_next(XBaseCommandScan * cmd, XBaseServer* tablet)
{	
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if (cmd == NULL || tablet == NULL || cmd->eof()) {
        return XERROR_SUCC;
    }

    cmd->next();

    execute(cmd, tablet);

    cmd->affected_rows(cmd->record().size());
    if ( tablet != mRootTablet 
        && cmd->eof()
        && !tablet->tablet_info().end_row().empty()
        && ( cmd->end_row().empty() ||  tablet->tablet_info().end_row() < cmd->end_row()) )
    {

        cmd->affected_rows(0);
        scan_close(cmd);
        if (cmd->fail()) {
            return cmd->result();
        }

        cmd->row() = tablet->tablet_info().end_row();
        scan_open(cmd);
        if ( cmd->eof() ) {
            return XERROR_SUCC;
        }
    }

    if ( tablet != mRootTablet && cmd->fatal() ) {
        XLOG4CPLUS_DEBUG(logger, "remove tablet");
        mTabletList->remove(*tablet);
    }

    return cmd->result();	
}


int XBaseClient::execute(XBaseCommandBase* cmd, XBaseServer* server)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if ( cmd == NULL || server == NULL ) {
        return XERROR_SUCC;
    }

    int encode_size = 0;
    for (;;) {
        encode_size = cmd->encode_request(mBuffer, MAX_BUFFER_SIZE);
        if (encode_size == XERROR_FAIL) {
            delete[] mBuffer;
            MAX_BUFFER_SIZE += MAX_BUFFER_SIZE;
            mBuffer = new char[MAX_BUFFER_SIZE];
            continue;
        } else if(encode_size == XERROR_CANCEL){
            return XERROR_SUCC;
        }else
            break;
    }

    int decode_size = server->execute(mBuffer, encode_size, MAX_BUFFER_SIZE);	
    if (decode_size < 0) {
        cmd->errorno( server->errorno() );
        if ( cmd->errorno() == XERROR_SERVER_CONNECT
            || cmd->errorno() == XERROR_SERVER_SEND
            || cmd->errorno() == XERROR_SERVER_RECV )
        {

            cmd->errorstr( server->errorstr() );			
        }
        cmd->result( XERROR_FAIL );
        XLOG4CPLUS_DEBUG( logger, cmd->to_string() );		
        return XERROR_FAIL;
    }

    int result = cmd->decode_response(mBuffer, decode_size);

    XLOG4CPLUS_DEBUG( logger, cmd->to_string() );
    return result;
}
int XBaseClient::execute(XBaseCommandBase* cmd)
{	
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if (cmd == NULL) {
        return XERROR_SUCC;
    }
    //DoZerg
    switch(cmd->_request_head->cmd_type()){
        case WRITE_REQ:
            return process_cmd_insert(dynamic_cast<XBaseCommandInsert *>(cmd));
        case READ_REQ:
            return process_cmd_select(dynamic_cast<XBaseCommandSelect *>(cmd));
        case TABLE_INFO_REQ:
            return process_cmd_desc(dynamic_cast<XBaseCommandDesc *>(cmd));
        case DELETE_TABLE_REQ:
            return process_cmd_drop(dynamic_cast<XBaseCommandDrop *>(cmd));
        case CREATE_TABLE_REQ:
            return process_cmd_create(dynamic_cast<XBaseCommandCreate *>(cmd));
        case 0:
        case LIST_TABLE_REQ:{
            XBaseServer* master = get_master();
            if (master == NULL) {
                cmd->errorno( mErrno );
                cmd->errorstr( mErrstr );
                return XERROR_FAIL;
            }
            return execute(cmd, master);}
        case BATCH_WRITE_REQ:
            return process_cmd_batch_insert(dynamic_cast<XBaseCommandBatchInsert *>(cmd));
        default:
            cmd->errorno( XERROR_MAX+1 );
            cmd->errorstr( "Unknown command" );
    }
    return XERROR_FAIL;
}

XBaseServer* XBaseClient::load_tablet(const  std::string & table_name, const  std::string & key)
{
    load_tablet(table_name);

    XBaseServer* server = mTabletList->get(table_name, key);
    if ( server == NULL ) {
        mErrno = mTabletList->errorno();
    }

    return server;
}

XBaseServer* XBaseClient::load_root_tablet()
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    XBaseCommandRootTablet cmd;	
    XBaseServer* master = get_master();
    if (master == NULL) {
        return NULL;
    }
    execute( &cmd, master );	
    if (cmd.fail()) {
        return NULL;
    }

    if (cmd.tablet_address().host_name().empty() || cmd.tablet_address().port() == 0) {
        mErrno = XERROR_ROOT_NOT_EXIST;
        return NULL;
    }

    mRootTablet->tablet_info() = cmd.tablet_info();
    mRootTablet->tabletAddress() = cmd.tablet_address();
    mRootTablet->init(cmd.tablet_address().host_name(), cmd.tablet_address().port(), mRecvTimeoutMs, mSendTimeoutMs);

    return mRootTablet;
}

bool XBaseClient::load_tablet(const std::string& table_name)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    // get the root tablet first
    XBaseServer* root_tablet = get_root_tablet();
    if (root_tablet == NULL) {
        return false;
    }

    root_tablet->row_from() = table_name;
    root_tablet->row_from() += ",";
    root_tablet->row_to() = table_name;
    root_tablet->row_to() += "-";

    XLOG4CPLUS_DEBUG( logger, "tablet_name=" << root_tablet->tablet_name() );
    XBaseCommandScan cmd(root_tablet->tablet_name(), root_tablet->row_from(), root_tablet->row_to(), "info:*");
    root_tablet->row_from() = "";
    root_tablet->row_to() = "";
    cmd.set_num_row( 1 );
    // open a scaner
    scan_open( &cmd, get_root_tablet() );

    if ( cmd.result() != PERR_SUCCESS && cmd.result() != PERR_END_OF_SCAN_RECORD) {
        root_tablet->close();
        root_tablet = load_root_tablet();
        if ( root_tablet == NULL ) {
            return false;
        }

        root_tablet->row_from() = table_name;
        root_tablet->row_from() += ",";
        root_tablet->row_to() = table_name;
        root_tablet->row_to() += "-";

        XLOG4CPLUS_DEBUG( logger, "tablet_name=" << root_tablet->tablet_name() );		

        scan_open( &cmd, get_root_tablet() );
    }
    if ( cmd.eof() ) {
        mErrno = PERR_TABLE_NOT_EXISTS;
        return false;
    }
    if (cmd.fail()) {
        mErrno = cmd.errorno();
        return false;
    }
    //DoZerg : delete old tablets
    mTabletList->remove(table_name);

    // iterate a scaner
    bool circle = true;
    bool succ = true;
    while (circle) {
        // next a scaner
        scan_next( &cmd, get_root_tablet() );
        if ( cmd.eof() && cmd.record().empty() ) {
            break;
        }
        if (cmd.fail()) {
            succ = false;
            break;
        }
        else {
            ServerAddress tablet_address;
            TabletInfo tablet_info;
            bool has_server_address = false, has_tablet_info = false;
            XBaseRecord& record = cmd.record();
            //iterate record(s)
            for (XBaseRecord::iterator it = record.begin(); it != record.end(); ++it) {
                if (it->key().row().find(table_name) != std::string::npos){
                    if (!has_server_address 
                        && !it->value().empty() 
                        && it->key().column().find( ROOT_META_COLUMN_NAME[RMC_SERVER_ADDRESS] ) == 0)
                    {
                        XStringStream decoder(it->value().c_str(), it->value().length(), XStringStream::NODELETE, XStringStream::BIGENDIAN);
                        if (!(decoder >> tablet_address)) { // buffer underflow
                            mErrno = XERROR_BUFFER_DECODE_UNDERFLOW;
                            circle = false;
                            succ = false;
                            break;
                        }
                        has_server_address = true;
                    }
                    else if (!has_tablet_info 
                        && !it->value().empty() 
                        && it->key().column().find( ROOT_META_COLUMN_NAME[RMC_TABLET_INFO] ) == 0)
                    {
                        XStringStream decoder(it->value().c_str(), it->value().length(), XStringStream::NODELETE, XStringStream::BIGENDIAN);					
                        if (!(decoder >> tablet_info)) { // buffer underflow
                            mErrno = XERROR_BUFFER_DECODE_UNDERFLOW;
                            circle = false;
                            succ = false;
                            break;
                        }
                        has_tablet_info = true;
                    }

                    XLOG4CPLUS_DEBUG( logger, "has_tablet_info=" << has_tablet_info << ",tablet_info.valid()=" << tablet_info.valid() );
                    if (has_tablet_info && tablet_info.valid() ) {
                        XBaseServer tablet;
                        tablet.tablet_info() = tablet_info;
                        tablet.tabletAddress() = tablet_address;
                        tablet.init(tablet_address.host_name(), tablet_address.port(), mRecvTimeoutMs, mSendTimeoutMs);

                        mTabletList->put(tablet);

                        has_server_address = false;
                        has_tablet_info = false;
                    }
                }
            }							
        }		

        if (cmd.eof()) {
            break;
        }
    }
    // close a scaner
    scan_close( &cmd, get_root_tablet() );
    //XLOG4CPLUS_DEBUG(logger, cmd.to_string());

    if (!succ || cmd.fail()) {
        return false;
    }
    return true;
}

int XBaseClient::turn_on_log()
{
#ifdef XLOGGER
    string logConfigFile = "xbase.log4cplus.properties";
    struct stat statBuf;
    int ret = stat( logConfigFile.c_str(), &statBuf );
    if ( ret == -1 ) {
        if ( errno == ENOENT ) { // Create the properties file
            ofstream of( logConfigFile.c_str() );
            if ( !of.good() ) {
                return -1;
            }

            of << "log4cplus.rootLogger=DEBUG, R" << endl;
            of << "log4cplus.appender.R=log4cplus::RollingFileAppender" << endl;
            of << "log4cplus.appender.R.File=xbase_client.log" << endl;
            of << "log4cplus.appender.R.MaxFileSize=50MB" << endl;
            of << "log4cplus.appender.R.MaxBackupIndex=10" << endl;
            of << "log4cplus.appender.R.layout=log4cplus::PatternLayout" << endl;
            of << "log4cplus.appender.R.layout.ConversionPattern=%d{%m/%d/%y %H:%M:%S} [%t] %-5p %c{2} %%%x%% - %m [%l]%n" << endl;
            of << "log4cplus.logger.filelogger = DEBUG" << endl;
        } else {
            return -1;
        }
    }
    log4cplus::PropertyConfigurator::doConfigure( logConfigFile.c_str() );
#endif

    return 0;
}


int XBaseClient::process_cmd_create(XBaseCommandCreate* cmd)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if ( cmd == NULL ) {
        return XERROR_SUCC;
    }

    if ( !(cmd->is_syntax_valid()) ) {
        cmd->errorno( PERR_INVALID_TABLE_STRUCT );
        return XERROR_FAIL;
    }

    XBaseServer* master = get_master();
    if (master == NULL) {
        cmd->errorno( mErrno );
        cmd->errorstr( mErrstr );

        return XERROR_FAIL;
    }

    return execute(cmd, master);
}
int XBaseClient::process_cmd_insert(XBaseCommandInsert* cmd)
{		
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );

    if ( cmd == NULL ) {
        return XERROR_SUCC;
    }

    if ( cmd->record().size() > 1 ) {
        XBaseRecord& record = cmd->record();		
        XBaseRecord::iterator it = record.begin();
        string& row = it->row();
        ++it;
        for ( ; it != record.end(); ++it ) {
            if ( it->row() != row ) {
                break;
            }
        }
        if ( it != record.end() ) {
            cmd->result( XERROR_FAIL );
            cmd->errorno( XERROR_BUFFER_DECODE_UNDERFLOW + 1 );
            cmd->errorstr( "You can't insert multi-row" );
            return XERROR_FAIL;
        }
    }

    XBaseServer* tablet = get_tablet(cmd);
    if (tablet == NULL) {
        return XERROR_FAIL;
    }	

    execute(cmd, tablet);
    if(cmd->fatal()) {
        XLOG4CPLUS_DEBUG(logger, "remove tablet");
        mTabletList->remove(*tablet);
    }

    return cmd->result();
}

int XBaseClient::process_cmd_batch_insert(XBaseCommandBatchInsert * cmd)
{
    XLOG4CPLUS_GUARD(logger,__FUNCTION__);
    if(cmd == NULL || !cmd->row_count())
        return XERROR_SUCC;
    std::string table = cmd->table_name();
    for(std::string start = cmd->first_row();;){
        XBaseServer * tablet = get_tablet(table,start);
        if(!tablet){
            XLOG4CPLUS_DEBUG( logger, "get_tablet() NULL: errno=" << mErrno << ": " << mErrstr );
            cmd->result( XERROR_FAIL );
            cmd->errorno( mErrno );
            cmd->errorstr( mErrstr );
            return XERROR_FAIL;
        }
        cmd->tablet_name() = tablet->tablet_name();
        std::string end = tablet->row_to();
        cmd->set_row_range(start,end);
        execute(cmd,tablet);
        if(cmd->fatal()) {
            XLOG4CPLUS_DEBUG(logger, "remove tablet");
            mTabletList->remove(*tablet);
            return cmd->result();
        }else if(!cmd->left())
            return cmd->result();
        start = end;
    }
    return cmd->result();
}

int XBaseClient::process_cmd_select(XBaseCommandSelect* cmd)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );
    if ( cmd == NULL ) {
        return XERROR_SUCC;
    }

    XBaseServer* tablet = get_tablet(cmd);
    if (tablet == NULL) {
        return XERROR_FAIL;
    }

    execute(cmd, tablet);
    if (cmd->fatal()) {
        XLOG4CPLUS_DEBUG(logger, "remove tablet");
        mTabletList->remove(*tablet);
    } else {
        cmd->affected_rows( cmd->record().size() );
    }

    return cmd->result();
}

int XBaseClient::process_cmd_desc(XBaseCommandDesc* cmd)
{	
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );
    if ( cmd == NULL ) {
        return XERROR_SUCC;
    }

    XBaseServer* master = get_master();
    if (master == NULL) {
        cmd->errorno( mErrno );
        cmd->errorstr( mErrstr );

        return XERROR_FAIL;
    }

    execute(cmd, master);
    if (!cmd->fail()) {
        cmd->affected_rows(cmd->family().size());
    }

    return cmd->result();
}

int XBaseClient::process_cmd_drop(XBaseCommandDrop* cmd)
{
    XLOG4CPLUS_GUARD( logger, __FUNCTION__ );
    if ( cmd == NULL ) {
        return XERROR_SUCC;
    }

    XBaseServer* master = get_master();
    if (master == NULL) {
        cmd->errorno( mErrno );
        cmd->errorstr( mErrstr );

        return XERROR_FAIL;
    }

    int ret = execute(cmd, master);
    if ( !cmd->fail() ) {
        mTabletList->remove( cmd->table_name() );
    }

    return ret;
}

